package org.zoomdev.zoom.hash;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.LongAdder;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.stream.Collectors;

public class ProductionArrayConsistentHash {
    // 读写锁保证线程安全
    private final ReentrantReadWriteLock rwLock = new ReentrantReadWriteLock();
    // 物理节点管理
    private final Set<String> physicalNodes = ConcurrentHashMap.newKeySet();
    private final int virtualNodesPerShard;
    // 统计信息
    private final LongAdder totalRequests = new LongAdder();
    private final Map<String, LongAdder> nodeCounters = new ConcurrentHashMap<>();
    private volatile HashRing currentHashRing = new HashRing(new int[0], new String[0]);

    public ProductionArrayConsistentHash() {
        this(160); // 默认每个节点160个虚拟节点
    }

    public ProductionArrayConsistentHash(int virtualNodesPerShard) {
        this.virtualNodesPerShard = virtualNodesPerShard;
    }

    /**
     * 添加物理节点（线程安全）
     */
    public void addNode(String node) {
        physicalNodes.add(node);
        rebuildHashRing();
    }

    /**
     * 移除物理节点（线程安全）
     */
    public void removeNode(String node) {
        physicalNodes.remove(node);
        rebuildHashRing();
    }

    /**
     * 查询节点（每秒支持10万+次查询）
     */
    public String getNode(String key) {
        HashRing snapshot = currentHashRing; // volatile读保证可见性
        int hash = hash(key);

        // 二分查找逻辑
        int idx = Arrays.binarySearch(snapshot.hashValues, hash);
        if (idx < 0) {
            idx = -idx - 1;
            if (idx >= snapshot.virtualNodeCount) idx = 0;
        }

        String node = snapshot.nodeNames[idx];
        recordStatistics(node);
        return node;
    }

    /**
     * 获取负载统计
     */
    public SystemStats getSystemStats() {
        return new SystemStats(
                totalRequests.sum(),
                currentHashRing.virtualNodeCount,
                physicalNodes.size(),
                new HashMap<>(nodeCounters.entrySet()
                        .stream()
                        .collect(Collectors.toMap(Map.Entry::getKey, e -> e.getValue().sum())))
        );
    }

    /**
     * 动态重建哈希环（带写锁保护）
     */
    private void rebuildHashRing() {
        rwLock.writeLock().lock();
        try {
            List<HashNode> nodes = generateVirtualNodes();
            HashRing newRing = buildSortedHashRing(nodes);
            currentHashRing = newRing; // volatile写保证可见性
        } finally {
            rwLock.writeLock().unlock();
        }
    }

    /**
     * 生成虚拟节点（可并行优化）
     */
    private List<HashNode> generateVirtualNodes() {
        List<HashNode> nodes = new ArrayList<>(physicalNodes.size() * virtualNodesPerShard);
        for (String node : physicalNodes) {
            for (int i = 0; i < virtualNodesPerShard; i++) {
                String vNode = node + "#VN" + i;
                nodes.add(new HashNode(hash(vNode), node));
            }
        }
        return nodes;
    }

    /**
     * 构建排序后的哈希环
     */
    private HashRing buildSortedHashRing(List<HashNode> nodes) {
        nodes.sort(Comparator.comparingInt(h -> h.hash));

        int[] hashValues = new int[nodes.size()];
        String[] nodeNames = new String[nodes.size()];

        for (int i = 0; i < nodes.size(); i++) {
            HashNode hn = nodes.get(i);
            hashValues[i] = hn.hash;
            nodeNames[i] = hn.node;
        }

        return new HashRing(hashValues, nodeNames);
    }

    /**
     * 高性能FNV1_32哈希算法
     */
    private static int hash(String key) {
        final int FNV_32_INIT = 0x811c9dc5;
        final int FNV_32_PRIME = 0x01000193;
        int hash = FNV_32_INIT;
        for (byte b : key.getBytes()) {
            hash ^= b;
            hash *= FNV_32_PRIME;
        }
        return hash & 0x7fffffff;
    }

    /**
     * 记录统计信息
     */
    private void recordStatistics(String node) {
        totalRequests.increment();
        nodeCounters.computeIfAbsent(node, k -> new LongAdder()).increment();
    }

    // 哈希环数据结构（不可变）
    private static class HashRing {
        final int[] hashValues;
        final String[] nodeNames;
        final int virtualNodeCount;

        HashRing(int[] hashValues, String[] nodeNames) {
            this.hashValues = hashValues;
            this.nodeNames = nodeNames;
            this.virtualNodeCount = hashValues.length;
        }
    }

    // 内部数据结构
    private static class HashNode {
        final int hash;
        final String node;

        HashNode(int hash, String node) {
            this.hash = hash;
            this.node = node;
        }
    }

    // 统计信息对象
    public static class SystemStats {
        public final long totalRequests;
        public final int virtualNodes;
        public final int physicalNodes;
        public final Map<String, Long> nodeDistribution;

        public SystemStats(long totalRequests, int virtualNodes,
                           int physicalNodes, Map<String, Long> nodeDistribution) {
            this.totalRequests = totalRequests;
            this.virtualNodes = virtualNodes;
            this.physicalNodes = physicalNodes;
            this.nodeDistribution = Collections.unmodifiableMap(nodeDistribution);
        }
    }
}